package com.atguigu.gulimall.order.config;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;

/**
 *
 */
@Configuration
public class MyRabbitConfig {

    //@Autowired
    RabbitTemplate rabbitTemplate;

    @Bean
    public MessageConverter messageConverter(){

        return new Jackson2JsonMessageConverter();
    }

    @Primary
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        this.rabbitTemplate = rabbitTemplate;
        rabbitTemplate.setMessageConverter(messageConverter());
        initRabbitTemplate();
        return rabbitTemplate;
    }

    /**
     * 定制RabbitTemplate
     * 1.RabbitMQ服务收到消息回调
     *   1)、#开启发送端确认 spring.rabbitmq.publisher-confirms=true
     *   2）、设置确认回调 ConfirmCallback
     * 2、消息正确抵达队列回调
     *   1)、#开启发送端消息抵达队列确认
     *       spring.rabbitmq.publisher-returns=true
     *       #只要抵达队列，以异步模式优先回调我们这个returnsconfirm
     *       spring.rabbitmq.template.mandatory=true
     *   2）、设置确认回调 ReturnCallback
     *
     *  3、消费端确认（保证每个消息正确消费，此时broker才可以删除消息）
     *     #手动ack消息
     *     spring.rabbitmq.listener.simple.acknowledge-mode=manual    手动签收
     *   1)、默认是自动确认，只要消息收到，客户端会自动确认，服务端就会移除这个消息
     *       问题 ：收到很多消息，自动回复给服务器ack，只有一个消息成功处理，然后宕机了，消息就丢失了，
     *       解决：手动确认
     *       #手动ack消息
     *       spring.rabbitmq.listener.simple.acknowledge-mode=manual
     *       消费者手动确认 :只要没有明确告诉MQ，货物被签收，没有ack，就一直是unacked状态。即使consumer宕机，会重
     *       新变为read状态，下一次新的consumer进来会发给他，（不会丢消息）
     *   2）、如何签收
     *       channel.basicAck(deliveryTag,false); 签收获取,业成功完成就应该签收
     *       channel.basicNack(deliveryTag,false,true);拒签：业务失败，拒签交给别人处理
     */

    //@PostConstruct  //MyRabbitConfig对象创建完以后，再来执行这个方法
    public void initRabbitTemplate(){
        //设置确认回调
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            /**
             * 只要消息抵达Broker ack就为true
             * @param correlationData  当前消息唯一关联数据（消息的唯一id）
             * @param ack              消息是否成功收到
             * @param cause            失败的原因
             */
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                //消息可靠性
                /** 防止消息丢失
                 * 1.做好消息确认机制(publisher,consumer[手动ACK])
                 * 2.每一个发送的消息都在数据库做好记录，定期将失败的消息再次发送一遍
                 *
                 * 防止消息重复
                 * 1.消息消费成功，事己经提交，ack时，机器宕机。导致没有ack成功，Broker消息重由unack 变为ready，并发
                 * 达给其他消费者
                 * 2.消息消费失败，由于重试机制，自动又将消息发送出去
                 * 3.成功消息，ack时宕机了，消息由unack变为ready，Broker又重新发送
                 *   1）消费者的业务消费接口应该设计为幂等性的，比如扣库存有工作单的状态标志
                 *   2）使用防重表(redis/mysql),发送消息每一个都有业务的唯一标识，处理过的不用处理
                 *   3）rabbitMQ的每一个消息都有redelivered字段，可获了是否被重新投递过来的，而不是第一次递过来的
                 *
                 * 消息积压
                 * 1.消费者宕机积压，
                 * 2.消费者消费力不足积压
                 * 3.发送者发送流量太大
                 *    1）上线更多的消息者，进行正常消费
                 *    2）上线专门的队列消费服务，将消息换批量取出来，记录数据库，离线慢慢处理
                 *
                 *
                 */
                //服务器收到了
                System.out.println("confirm....["+correlationData+"]==>ack["+ack+"]"
                        + "==>cause["+cause+"]");
            }
        });

        //设置消息抵达队列的确认回调,
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            /**
             * 只要消息没有投递给指定的队列，就触发这个失败回调
             * @param message     投递失败消息的详细信息
             * @param replyCode   回复的状态码
             * @param replyText   回复的文本内容
             * @param exchange    当时消息发给哪个交换机
             * @param routingKey  当时消息用的哪个路由键
             */
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                //报错了，修改数据库当前消息的状态->错误
//                System.out.println("Fail Message[" +message+ "]" +
//                        "==>replyCode["+replyCode+"]" +
//                        "==>replyText["+replyText+"]" +
//                        "==>exchange["+exchange+"]" +
//                        "==>routingKey["+routingKey+"]"
//                );
            }
        });
    }

}
